package top.jacktgq.concurrent_utils.pool;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import top.jacktgq.utils.LogUtils;
import top.jacktgq.utils.Sleeper;
/**
 * 
 * @Title: MyThreadPool.java 
 * @Package top.jacktgq.concurrent_utils 
 * @Description: 自定义一个线程池
 * @author CandyWall   
 * @date 2020年11月4日 上午1:46:17 
 * @version V1.0
 */
public class MyThreadPoolDemo {

	public static void main(String[] args) {
		MyThreadPool myThreadPool 
		     = new MyThreadPool(2, 1000, TimeUnit.MILLISECONDS, 5, (taskQueue, task) -> {
		    //1、死等
		    //taskQueue.put(task);
		    //2、超时等待，超过时间就放弃这个任务
			/*if(!taskQueue.offer(task, 1000, TimeUnit.MILLISECONDS)) {
				LogUtils.log("MyThreadPoolImproveBySemaphoreDemo", "超出任务等待时间，任务" + task + "被放弃...");
			}*/
			//3、直接放弃任务执行
	    	//LogUtils.log("MyThreadPoolImproveBySemaphoreDemo", "任务队列已满，任务 " + task + "直接被放弃...");
			//4、抛出异常
			//throw new RuntimeException("队列已经满了，任务无法添加到队列中，任务 " + task + " 执行失败！");
			//5、让调用者自己执行任务
		    //调用了主线程执行的 23333
		    task.run();
		});
		for(int i = 0; i < 15; i++) {
			int j = i + 1;
			myThreadPool.execute(() -> {
				Sleeper.sleep(1500, TimeUnit.MILLISECONDS);
				LogUtils.log("MyThreadPoolImproveBySemaphoreDemo", "" + j);
			});
		}
	}

}

@FunctionalInterface
//拒绝策略：支持用户自己处理任务队列满了的情况该怎么处理任务
interface RejectPolicy<T> {
	void reject(BlockingQueue<T> taskQueue, T task);
}

class MyThreadPool {
	//任务队列
	private BlockingQueue<Runnable> taskQueue;
	
	//线程集合
	private HashSet<Worker> workers = new HashSet<Worker>();
	
	//核心线程数
	private int coreSize;
	
	//获取任务的超时时间
	private long timeout;
	
	//超时时间的单位
	private TimeUnit timeUnit;
	
	//拒绝策略
	
	private RejectPolicy<Runnable> rejectPolicy;
	
	private class Worker extends Thread {
		private Runnable task;
		public Worker(Runnable task) {
			this.task = task;
		}
		
		@Override
		public void run() {
			//执行任务
			//1、当task不为空，执行任务
			//2、当task执行完毕，再接着从任务队列中获取任务并执行
			//while(task != null || (task = taskQueue.take()) != null) {
			while(task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
				try {
					LogUtils.log("MyThreadPool", "正在执行 " + task +" 任务");
					task.run();
				} catch(Exception e) {
					e.printStackTrace();
				} finally {
					task = null;
				}
			}
			synchronized (workers) {
				LogUtils.log("MyThreadPool", "worker " + this.getName() +" 被移除");
				//从当前线程池中移除该线程
				workers.remove(this);
			}
		}
	}
	
	public MyThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapcity, RejectPolicy<Runnable> rejectPolicy) {
		this.coreSize = coreSize;
		this.timeout = timeout;
		this.timeUnit = timeUnit;
		this.taskQueue = new BlockingQueue<Runnable>(queueCapcity);
		this.rejectPolicy = rejectPolicy;
	}
	
	public void execute(Runnable task) {
		synchronized (workers) {
			//当任务数没有超过核心coreSize时，直接交给worker对象执行
			if (workers.size() < coreSize) {
				Worker worker = new Worker(task);
				LogUtils.log("MyThreadPool", "新增Worker：" + worker.getName() + "，Task：" + task);
				workers.add(worker);
				worker.start();
			}
			//如果任务数超过核心coreSize时，加入任务队列暂存
			else {
				//如果队列满了有以下可以做的操作：如果这里选择了某一种写死不太好，这里采用策略模式，让用户自己选择
				taskQueue.tryPut(rejectPolicy, task);
			}
		}
	}
}

class BlockingQueue<T> {
	//1、任务队列（Deque是一个双向队列）
	private Deque<T> queue = new ArrayDeque();
	
	//2、锁
	ReentrantLock lock = new ReentrantLock();
	
	//3、生产者条件变量，队列容量到达上限就让生产者线程进去等待
	private Condition fullWaitSet = lock.newCondition();
	
	//4、消费者条件变量，队列中没有元素时就让消费者线程进去等待
	private Condition emptyWaitSet = lock.newCondition();
	
	//5、容量
	private int capacity;

	public BlockingQueue(int capacity) {
		this.capacity = capacity;
	}
	
	//带超时的阻塞获取
	public T poll(long timeout, TimeUnit timeUnit) {
		lock.lock();
		try {
			//将Timeout统一转换成纳秒
			long nanos = timeUnit.toNanos(timeout);
			//任务队列为空
			while(queue.isEmpty()) {
				try {
					//如果剩余的等待时间小于等于0，就直接返回空任务
					if(nanos <= 0) {
						return null;
					}
					//返回的是剩余时间
					nanos = emptyWaitSet.awaitNanos(nanos);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWaitSet.signalAll();
			return t;
		} finally {
			lock.unlock();
		}
	}
	
	//阻塞获取
	public T take() {
		lock.lock();
		try {
			//任务队列为空
			while(queue.isEmpty()) {
				try {
					emptyWaitSet.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T task = queue.removeFirst();
			fullWaitSet.signalAll();
			return task;
		} finally {
			lock.unlock();
		}
	}
	
	//采用策略模式，用户可以自定义任务队列满了的情况怎么处理任务
	public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
		lock.lock();
		try {
			//判断队列是否已满
			if(queue.size() == capacity) {
				//1、死等
				//2、调用带超时的等待
				//3、放弃任务执行
				//4、抛出异常
				//5、让调用者自己执行任务
				rejectPolicy.reject(this, task);
			} 
			//有空闲
			else {
				LogUtils.log("MyThreadPool", "任务 " + task +" 加入队列");
				queue.addLast(task);
				emptyWaitSet.signalAll();
			}
		} finally {
			lock.unlock();
		}
	}
	
	//带超时的阻塞添加
	public boolean offer(T task, long timeout, TimeUnit timeUnit) {
		lock.lock();
		try {
			//将Timeout统一转换成纳秒
			long nanos = timeUnit.toNanos(timeout);
			//任务队列满了
			while(queue.size() == capacity) {
				try {
					LogUtils.log("MyThreadPool", "任务 " + task +" 等待加入队列");
					//如果剩余的等待时间小于等于0，就直接返回空任务
					if(nanos <= 0) {
						return false;
					}
					//返回的是剩余时间
					nanos = fullWaitSet.awaitNanos(nanos);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			LogUtils.log("MyThreadPool", "任务 " + task +" 加入队列");
			queue.addLast(task);
			emptyWaitSet.signalAll();
			return true;
		} finally {
			lock.unlock();
		}
	}
	//阻塞添加
	public void put(T task) {
		lock.lock();
		try {
			//任务队列满了
			while(queue.size() == capacity) {
				try {
					LogUtils.log("MyThreadPool", "任务 " + task +" 等待加入队列");
					fullWaitSet.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			LogUtils.log("MyThreadPool", "任务 " + task +" 加入队列");
			queue.addLast(task);
			emptyWaitSet.signalAll();
		} finally {
			lock.unlock();
		}
	}
	
	//获取大小
	public int size() {
		lock.lock();
		try {
			return queue.size();
		} finally {
			lock.unlock();
		}
	}
}
